package com.example.dataextraction.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.example.dataextraction.model.*;
import com.example.dataextraction.model.vo.CopyTableVO;
import com.example.dataextraction.utils.DBHelper;
import org.reflections.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.mail.javamail.MimeMessageHelper;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.mail.MessagingException;
import javax.mail.internet.MimeMessage;
import java.sql.*;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.Date;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static com.example.dataextraction.enums.DataSourceEnum.DEFAULT_DATA_SOURCE;

/**
 * @author Zenan Chen
 * @date 2023/1/16 14:29
 */
@Service
//@DependsOn("defaultDataSource")
public class FetchDataServiceImpl implements FetchDataService {
    private static final Logger log = LoggerFactory.getLogger(FetchDataServiceImpl.class);

    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    JavaMailSender javaMailSender;

    /**
     * 创建数据库
     *
     * @param dataBaseName 数据库名
     * @throws Exception
     */
    @Override
    public void createDataBase(String dataSourceCode, String dataBaseName) throws Exception {
        DBHelper dbHelper = getDbHelper(dataSourceCode, "");
        QueryVO queryVO = new QueryVO();
        queryVO.setSql("show databases;");
        List<JSONObject> result = getRS2JsonObject(dbHelper, queryVO);
        List<String> databases = new ArrayList<>();
        if (DEFAULT_DATA_SOURCE.getCode().equals(dataSourceCode)) {
            databases = result.stream().map(jsonObject -> {
                JSONArray schema_name = jsonObject.getJSONArray("Database");
                String databaseName = schema_name.getString(1);
                return databaseName;
            }).collect(Collectors.toList());
        } else {
            databases = result.stream().map(jsonObject -> {
                JSONArray schema_name = jsonObject.getJSONArray("SCHEMA_NAME");
                String databaseName = schema_name.getString(1);
                return databaseName;
            }).collect(Collectors.toList());
        }
        if (databases != null && databases.size() > 0 && !databases.contains(dataBaseName)) {
            String sql = "create database " + dataBaseName;
            dbHelper.executeSql(sql, true);
            log.info("---------------------> 创建数据库：【{}】，成功！ ", dataBaseName);
        }
    }

    /**
     * 根据数据源类型初始化数据源
     *
     * @param dataSourceCode
     * @param dataBaseName
     * @return
     * @throws Exception
     */
    private DBHelper getDbHelper(String dataSourceCode, String dataBaseName) throws Exception {
        DataSource dataSource = dataSourceService.getByCode(dataSourceCode);
        if (!Utils.isEmpty(dataBaseName)) {
            log.info("---------------------> before  dataSource:【{}】", JSON.toJSONString(dataSource));
            String port = dataSource.getPort();
            String url = dataSource.getUrl();
            String subUrl = url.substring(0, url.lastIndexOf(":"));
            String paramStr = "";
            if (url.contains("?")) {
                paramStr = url.substring(url.indexOf("?"));
            }
            dataSource.setUrl(subUrl + ":" + port + "/" + dataBaseName + paramStr);
            log.info("---------------------> after  dataSource:【{}】", JSON.toJSONString(dataSource));
        }
        return new DBHelper(dataSource);
    }


    /**
     * 复制数据库表
     *
     * @param copyTableVO
     * @throws Exception
     */
    @Override
    public void copyTable(CopyTableVO copyTableVO) throws Exception {
        String fromDataBaseName = copyTableVO.getFromDataBaseName();
        String fromTableName = copyTableVO.getFromTableName();
        String toDataBaseName = copyTableVO.getToDataBaseName();
        String toTableName = copyTableVO.getToTableName();
        // 初始化【来源数据源】,
        DBHelper fromDBHelper = getDbHelper(copyTableVO.getFromDataSourceCode(), copyTableVO.getFromDataBaseName());
        // 先校验【目标数据库】是否存在
        createDataBase(copyTableVO.getToDataSourceCode(), toDataBaseName);
        // 校验【目标数据库】对应的表是否存在
        boolean tableExist = checkTableExist(copyTableVO.getToDataSourceCode(), toDataBaseName, toTableName);
        if (tableExist) {
            throw new RuntimeException(toDataBaseName + "." + toTableName + "已经存在，无法复制！");
        }
        // 复制表
        QueryVO queryVO = new QueryVO();
        String sql = "show create table " + fromDataBaseName + "." + fromTableName;
        queryVO.setSql(sql);
        List<JSONObject> result = getRS2JsonObject(fromDBHelper, queryVO);
        for (JSONObject item : result) {
            if (!StringUtils.isEmpty(item.getString("Create Table"))) {
                String createTableSQL = item.getString("Create Table");
                createTableSQL = createTableSQL.substring(createTableSQL.indexOf("CREATE TABLE"));
                if (createTableSQL.endsWith("]")) {
                    createTableSQL = createTableSQL.substring(0, createTableSQL.length() - 1);
                }
                Pattern compile = Pattern.compile("`.*`");
                Matcher matcher = compile.matcher(createTableSQL);
                createTableSQL = matcher.replaceFirst("`" + toTableName + "`");
                DBHelper toDBHelper = getDbHelper(copyTableVO.getToDataSourceCode(), toDataBaseName);
                log.info("---------------------> 复制【{}.{}】的sql：【{}】！ ", fromDataBaseName, fromTableName, createTableSQL);
                try {
                    toDBHelper.executeSql(createTableSQL, true);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * 执行sql语句
     *
     * @param queryVO
     * @return
     * @throws Exception
     */
    @Override
    public Object executeSQL(QueryVO queryVO) throws Exception {
        DBHelper dbHelper = getDbHelper(queryVO.getDataSourceCode(), "");
        List<JSONObject> result = getRS2JsonObject(dbHelper, queryVO);
        return result;
    }


    /**
     * 校验数据库是否存在
     *
     * @param dbName 数据库名称
     * @return
     * @throws Exception
     */
    private boolean checkDataBaseExist(String dataSourceCode,String dbName) throws Exception {
        DBHelper dbHelper = getDbHelper(dataSourceCode, dbName);
        boolean dbExist = false;
        Connection connection = dbHelper.getConnection();
        DatabaseMetaData metaData = connection.getMetaData();
        ResultSet schemas = metaData.getSchemas(null, dbName);
        dbExist = schemas.next();
        return dbExist;
    }


    /**
     * 校验数据库下的表是否存在
     *
     * @param dbName
     * @param tblName
     * @return
     * @throws Exception
     */
    private boolean checkTableExist(String dataSourceCode, String dbName, String tblName) throws Exception {
        DBHelper dbHelper = getDbHelper(dataSourceCode, "");
        boolean tableExist = false;
        Connection connection = dbHelper.getConnection();
        DatabaseMetaData metaData = connection.getMetaData();
        ResultSet rs = metaData.getTables(dbName, null, tblName, new String[]{"TABLE"});
        tableExist = rs.next();
        return tableExist;
    }


    /**
     * 将查询语句转化为List<JsonObject>
     *
     * @param dbHelper
     * @param queryVO
     * @return
     * @throws Exception
     */
    private List<JSONObject> getRS2JsonObject(DBHelper dbHelper, QueryVO queryVO) throws Exception {
        ResultSet rs = null;
        List<JSONObject> result = new ArrayList<>();
        try {
            String sql = queryVO.getSql();
            if (null != queryVO.getLimit() && true == queryVO.getLimit()) {
                Integer pageNum = queryVO.getPageNum();
                Integer pageSize = queryVO.getPageSize();
                Integer start = (pageNum - 1) * pageSize;
                Integer end = pageSize;
                sql = "select * from (" + sql.replace(";", " ") + ") temp order by null limit " + start + "," + end;
            }
            log.info("---------------------> 执行sql：【{}】", sql);
            rs = dbHelper.executeQuery(sql, null);
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();
            while (rs.next()) {
                JSONObject jsonObject = new JSONObject();
                for (int i = 1; i <= columnCount; i++) {
                    String key = metaData.getColumnName(i);
                    int columnType = metaData.getColumnType(i);
                    String value = rs.getString(i);
                    ArrayList<Object> info = new ArrayList<>();
                    info.add(columnType);
                    info.add(value);
                    jsonObject.put(key, info);
                }
                result.add(jsonObject);
            }
        } catch (Exception e) {
            log.error("---------------------> 执行sql：【{}】，发生错误：【{}】", queryVO.getSql(), e.getStackTrace());
        } finally {
            dbHelper.close();
        }
        return result;
    }

    /**
     * 检测数据库状态
     *
     * @throws Exception
     */
    public void checkDorisStatus(String dataSourceCode) throws Exception {
        DBHelper dbHelper = getDbHelper(dataSourceCode, "");
        QueryVO queryVO = new QueryVO();
        String sql = "SHOW proc '/backends';";
        queryVO.setSql(sql);
        List<JSONObject> rs2JsonObject = getRS2JsonObject(dbHelper, queryVO);
        List<JSONObject> alive = rs2JsonObject.stream().filter(item -> {
            return item.getBoolean("Alive");
        }).collect(Collectors.toList());
        if (alive.size() < 3) {
            sendEmail();
            throw new RuntimeException("doris节点异常！");
        }
    }

    /**
     * 造数
     *
     * @throws Exception
     */
    @Override
    public void createData(String dataSourceCode) throws Exception {
        DBHelper dbHelper = getDbHelper(dataSourceCode, "");
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
        List<String> allDateList = new ArrayList<>();
        LocalDate startDate = LocalDate.of(2022, 1, 1);
        LocalDate endDate = startDate;
        while (2022 == endDate.getYear()) {
            allDateList.add(endDate.format(formatter));
            endDate = endDate.plusDays(1L);
        }
        for (String date : allDateList) {
            String sql = "insert into DIM.DIM_PERIOD_WID(PERIOD,PERIOD_TYPE,START_MONTH,END_MONTH) values('" + date + "','天',0,0)";
            dbHelper.executeSql(sql, true);
        }

    }


    /**
     * 将源表数据抽取到目标表
     *
     * @param copyTableVO
     * @throws Exception
     */
    @Override
    public void copyData(CopyTableVO copyTableVO) throws Exception {
        String fromDataBaseName = copyTableVO.getFromDataBaseName();
        String fromTableName = copyTableVO.getFromTableName();
        String toDataBaseName = copyTableVO.getToDataBaseName();
        String toTableName = copyTableVO.getToTableName();
        Boolean cover = copyTableVO.getCover();
        DBHelper fromDBHelper = getDbHelper(copyTableVO.getFromDataSourceCode(), "");
        // 1.检查源库表和目标库是否存在
        boolean srcTblExist = checkTableExist(copyTableVO.getFromDataSourceCode(), fromDataBaseName, fromTableName);
        if (!srcTblExist) {
            throw new RuntimeException(fromDataBaseName + "." + fromTableName + "不存在！");
        }
        // 2.如果目标表不存在，则自动创建，表名和源表名一致
        boolean tgtTblExist = checkTableExist(copyTableVO.getToDataSourceCode(), toDataBaseName, toTableName);
        if (!tgtTblExist) {
            copyTable(copyTableVO);
        }
        // 3.进行抽数
        // 3.1 计算分批抽数的批次
        QueryVO queryVO = new QueryVO();
        String countSQL = "select count(1) as total from " + fromDataBaseName + "." + fromTableName;
        queryVO.setSql(countSQL);
        queryVO.setLimit(false);
        List<JSONObject> countJSON = getRS2JsonObject(fromDBHelper, queryVO);
        Long total = countJSON.get(0).getJSONArray("total").getLong(1);
        Long batchs = (total / copyTableVO.getLimit()) + 1;
        log.info("---------------------> 【{}】共有【{}】条数据，需分【{}】批导入，每批【{}】条", fromDataBaseName + "." + fromTableName, total, batchs, copyTableVO.getLimit());
        DBHelper toDbHelper = getDbHelper(copyTableVO.getToDataSourceCode(), toDataBaseName);
        // 如果是覆盖，那么先truncate表再插入
        if (cover) {
            log.info("---------------------> cover属性为true，先清空表数据！");
            String truncateSQL = "truncate table " + toDataBaseName + "." + toTableName;
            toDbHelper.executeSql(truncateSQL, true);
            log.info("---------------------> 执行 truncateSQL：【{}】", truncateSQL);
        }
        for (int i = 0; i < batchs; i++) {
            log.info("---------------------> 第【{}】批导入开始", i + 1);
            // 3.2 将即将抽数的数据存放到内存中
            String querySQL = "select *" + " from " + fromDataBaseName + "." + fromTableName;
            queryVO.setLimit(true);
            queryVO.setSql(querySQL);
            queryVO.setPageNum(i + 1);
            queryVO.setPageSize(copyTableVO.getLimit());
            List<JSONObject> records = getRS2JsonObject(fromDBHelper, queryVO);
            StringBuilder colNameSQLSB = new StringBuilder();
            StringBuilder valuesSQLSB = new StringBuilder("values ");
            boolean firstRound = true;
            for (JSONObject record : records) {
                valuesSQLSB.append("(");
                for (String recordKey : record.keySet()) {
                    ArrayList list = record.getObject(recordKey, ArrayList.class);
                    Integer colType = (Integer) list.get(0);
                    String colValue = (String) list.get(1);
                    if (firstRound) {
                        colNameSQLSB.append("`").append(recordKey).append("`").append(",");
                    }
                    if (Types.VARCHAR == colType || Types.TIMESTAMP == colType || Types.LONGVARCHAR == colType || "DORIS".equals(copyTableVO.getToDataSourceCode())) {
                        if (null == colValue || "null".equals(colValue) || "NULL".equals(colValue)) {
                            valuesSQLSB.append("NULL").append(",");
                        } else {
                            valuesSQLSB.append("'").append(colValue).append("'").append(",");
                        }
                    } else {
                        if (null == colValue || "null".equals(colValue) || "NULL".equals(colValue)) {
                            valuesSQLSB.append("NULL").append(",");
                        } else {
                            valuesSQLSB.append(colValue).append(",");
                        }
                    }
                }
                valuesSQLSB = valuesSQLSB.deleteCharAt(valuesSQLSB.length() - 1);
                valuesSQLSB.append("),");
                firstRound = false;
            }
            String colNameSQL = colNameSQLSB.substring(0, colNameSQLSB.toString().length() - 1);
            String valuesSQL = valuesSQLSB.substring(0, valuesSQLSB.toString().length() - 1);
            // 3.2 将源表的数据抽取到目标表
            StringBuffer extractSQL = new StringBuffer();
            extractSQL.append("INSERT INTO ").append(toDataBaseName).append(".").append(toTableName).append("(").append(colNameSQL).append(") ").append(valuesSQL);
//            log.info("---------------------> 执行 extractSQL：【{}】", extractSQL);
            toDbHelper.executeSql(extractSQL.toString(), true);
        }
        log.info("---------------------> 从" + fromDataBaseName + "." + fromTableName + "抽数到" + toDataBaseName + "." + toTableName + "完成！");
    }

    @Override
    public List<JSONObject> getColumnInfo(String dataSourceCode, String dataBaseName, String tableName) throws Exception {
        DBHelper dbHelper = getDbHelper(dataSourceCode, dataBaseName);
        Connection connection = dbHelper.getConnection();
        DatabaseMetaData metaData = connection.getMetaData();
        ResultSet columnsRS = metaData.getColumns(dataBaseName, null, tableName, null);
        List<JSONObject> result = new ArrayList<>();
        JSONObject jsonObject = new JSONObject();
        int i = 1;
        while (columnsRS.next()) {
            log.info("【column{}】---------------------> TABLE_CAT：【{}】", i, columnsRS.getString("TABLE_CAT"));
            jsonObject.put("TABLE_CAT", columnsRS.getString("TABLE_CAT"));
            log.info("【column{}】---------------------> TABLE_SCHEM：【{}】", i, columnsRS.getString("TABLE_SCHEM"));
            jsonObject.put("TABLE_SCHEM", columnsRS.getString("TABLE_SCHEM"));
            log.info("【column{}】---------------------> TABLE_NAME：【{}】", i, columnsRS.getString("TABLE_NAME"));
            jsonObject.put("TABLE_NAME", columnsRS.getString("TABLE_NAME"));
            log.info("【column{}】---------------------> COLUMN_NAME：【{}】", i, columnsRS.getString("COLUMN_NAME"));
            jsonObject.put("COLUMN_NAME", columnsRS.getString("COLUMN_NAME"));
            log.info("【column{}】---------------------> DATA_TYPE：【{}】", i, columnsRS.getString("DATA_TYPE"));
            jsonObject.put("DATA_TYPE", columnsRS.getString("DATA_TYPE"));
            log.info("【column{}】---------------------> TYPE_NAME：【{}】", i, columnsRS.getString("TYPE_NAME"));
            jsonObject.put("TYPE_NAME", columnsRS.getString("TYPE_NAME"));
            log.info("【column{}】---------------------> COLUMN_SIZE：【{}】", i, columnsRS.getString("COLUMN_SIZE"));
            jsonObject.put("COLUMN_SIZE", columnsRS.getString("COLUMN_SIZE"));
            log.info("【column{}】---------------------> BUFFER_LENGTH：【{}】", i, columnsRS.getString("BUFFER_LENGTH"));
            jsonObject.put("BUFFER_LENGTH", columnsRS.getString("BUFFER_LENGTH"));
            log.info("【column{}】---------------------> DECIMAL_DIGITS：【{}】", i, columnsRS.getString("DECIMAL_DIGITS"));
            jsonObject.put("DECIMAL_DIGITS", columnsRS.getString("DECIMAL_DIGITS"));
            log.info("【column{}】---------------------> NUM_PREC_RADIX：【{}】", i, columnsRS.getString("NUM_PREC_RADIX"));
            jsonObject.put("NUM_PREC_RADIX", columnsRS.getString("NUM_PREC_RADIX"));
            log.info("【column{}】---------------------> NULLABLE：【{}】", i, columnsRS.getString("NULLABLE"));
            jsonObject.put("NULLABLE", columnsRS.getString("NULLABLE"));
            log.info("【column{}】---------------------> REMARKS：【{}】", i, columnsRS.getString("REMARKS"));
            jsonObject.put("REMARKS", columnsRS.getString("REMARKS"));
            log.info("【column{}】---------------------> COLUMN_DEF：【{}】", i, columnsRS.getString("COLUMN_DEF"));
            jsonObject.put("COLUMN_DEF", columnsRS.getString("COLUMN_DEF"));
            log.info("【column{}】---------------------> SQL_DATA_TYPE：【{}】", i, columnsRS.getString("SQL_DATA_TYPE"));
            jsonObject.put("SQL_DATA_TYPE", columnsRS.getString("SQL_DATA_TYPE"));
            log.info("【column{}】---------------------> SQL_DATETIME_SUB：【{}】", i, columnsRS.getString("SQL_DATETIME_SUB"));
            jsonObject.put("SQL_DATETIME_SUB", columnsRS.getString("SQL_DATETIME_SUB"));
            log.info("【column{}】---------------------> CHAR_OCTET_LENGTH：【{}】", i, columnsRS.getString("CHAR_OCTET_LENGTH"));
            jsonObject.put("CHAR_OCTET_LENGTH", columnsRS.getString("CHAR_OCTET_LENGTH"));
            log.info("【column{}】---------------------> ORDINAL_POSITION：【{}】", i, columnsRS.getString("ORDINAL_POSITION"));
            jsonObject.put("ORDINAL_POSITION", columnsRS.getString("ORDINAL_POSITION"));
            log.info("【column{}】---------------------> IS_NULLABLE：【{}】", i, columnsRS.getString("IS_NULLABLE"));
            jsonObject.put("IS_NULLABLE", columnsRS.getString("IS_NULLABLE"));
            log.info("【column{}】---------------------> SCOPE_CATALOG：【{}】", i, columnsRS.getString("SCOPE_CATALOG"));
            jsonObject.put("SCOPE_CATALOG", columnsRS.getString("SCOPE_CATALOG"));
            log.info("【column{}】---------------------> SCOPE_SCHEMA：【{}】", i, columnsRS.getString("SCOPE_SCHEMA"));
            jsonObject.put("SCOPE_SCHEMA", columnsRS.getString("SCOPE_SCHEMA"));
            log.info("【column{}】---------------------> SCOPE_TABLE：【{}】", i, columnsRS.getString("SCOPE_TABLE"));
            jsonObject.put("SCOPE_TABLE", columnsRS.getString("SCOPE_TABLE"));
            log.info("【column{}】---------------------> SOURCE_DATA_TYPE：【{}】", i, columnsRS.getString("SOURCE_DATA_TYPE"));
            jsonObject.put("SOURCE_DATA_TYPE", columnsRS.getString("SOURCE_DATA_TYPE"));
            log.info("【column{}】---------------------> IS_AUTOINCREMENT：【{}】", i, columnsRS.getString("IS_AUTOINCREMENT"));
            jsonObject.put("IS_AUTOINCREMENT", columnsRS.getString("IS_AUTOINCREMENT"));
            log.info("【column{}】---------------------> IS_GENERATEDCOLUMN：【{}】", i, columnsRS.getString("IS_GENERATEDCOLUMN"));
            jsonObject.put("IS_GENERATEDCOLUMN", columnsRS.getString("IS_GENERATEDCOLUMN"));
            result.add(jsonObject);
            i++;
        }
        return result;
    }

    /**
     * 发送邮件
     *
     * @throws MessagingException
     */
    private void sendEmail() throws MessagingException {
        MimeMessage mimeMessage = javaMailSender.createMimeMessage();
        MimeMessageHelper helper = new MimeMessageHelper(mimeMessage, true);
        helper.setSubject("Doris节点状态检测");
        helper.setFrom("chenzn15@meicloud.com");
        helper.setTo("chenzn15@meicloud.com");
        helper.setSentDate(new Date());
        helper.setText("Doris节点出现异常，请尽快查看！");
        //send the mail
        javaMailSender.send(mimeMessage);
    }

}
